import os
import re
import sys
import math
import random
import struct
import logging
import datetime
import talib as ta
import pandas as pd
import backtrader as bt
from multiprocessing import Pool
from mootdx.quotes import Quotes
from collections import namedtuple
from pandas import Timedelta
from pytdx.config.hosts import hq_hosts
from pytdx.hq import TdxHq_API
from pytdx.params import TDXParams
from pytdx.pool.hqpool import TdxHqPool_API
from pytdx.pool.ippool import AvailableIPPool
from pytdx.reader import TdxLCMinBarReader, TdxFileNotFoundException, BlockReader
from pytdx.reader.block_reader import BlockReader_TYPE_GROUP


logging.basicConfig(format="%(asctime)s %(filename)s(line:%(lineno)d) [%(levelname)s] : %(message)s", datefmt="%Y-%M-%d %H:%M:%S", level=logging.DEBUG)


def resolve(TdxPath, *args, **kwargs):
    return os.path.join(TdxPath, *args, **kwargs)

'''通过证券代码获取市场信息'''


def get_stock_market(code):
    patterns = (
        (0, '深圳指数', '深圳证券交易所', '^399\d{3}$', 'sz'),
        (1, '上证A股', '上海证券交易所', '^60[0135]\d{3}$', 'sh'),
        (0, '深证A股', '深圳证券交易所', '^00\d{4}$', 'sz'),
        (1, '上证B股', '上海证券交易所', '^900\d{3}$', 'sh'),
        (0, '深证B股', '深圳证券交易所', '^200\d{3}$', 'sz'),
        (2, '北证A股', '北京证券交易所', '^(43|83|87|88)\d{4}$', 'bj'),
        (1, '中小板', '深圳证券交易所', '^002\d{3}$', 'sz'),
        (1, '科创板', '上海证券交易所', '^68\d{4}$', 'sh'),
        (0, '创业板', '深圳证券交易所', '^30\d{4}$', 'sz'),
        (3, '港股', '香港证券交易所', '\d{4,5}.HK$', 'hk'),
        (1, '上证国债现货', '上海证券交易所', '^001\d{3}$', 'sh'),
        (1, '上证企业债券', '上海证券交易所', '^0[12]0\d{3}$', 'sh'),
        (1, '上证可转换债券', '上海证券交易所', '^1[20][90]\d{3}$', 'sh'),
        (1, '上证国债回购', '上海证券交易所', '^204\d{3}$', 'sh'),
        (1, '上证国债期货', '上海证券交易所', '^310\d{3}$', 'sh'),
        (0, '深证国债现货', '深圳证券交易所', '^10\d{4}$', 'sz'),
        (0, '深证企业债券', '深圳证券交易所', '^11\d{4}$', 'sz'),
        (0, '深证可转换债券', '深圳证券交易所', '^12\d{4}$', 'sz'),
        (0, '深证国债回购', '深圳证券交易所', '^131\d{3}$', 'sz'),
        (1, '上证封闭式基金', '上海证券交易所', '^508\d{3}$', 'sh'),
        (1, '上证ETF基金', '上海证券交易所', '^(51|56|58)\d{4}$', 'sh'),
        (1, '上证LOF基金', '上海证券交易所', '^501\d{3}$', 'sh'),
        (1, '上证分级基金', '上海证券交易所', '^(502|506)\d{3}$', 'sh'),
        (0, '深证封闭式基金', '深圳证券交易所', '^184\d{3}$', 'sz'),
        (0, '深证ETF基金', '深圳证券交易所', '^159\d{3}$', 'sz'),
        (0, '深证LOF基金', '深圳证券交易所', '^16\d{4}$', 'sz'),
        (0, '深证分级基金', '深圳证券交易所', '^15\d{4}$', 'sz'),
    )
    for m in patterns:
        if re.match(m[3], code):
            Market = namedtuple('Market', ['stock_code', 'market', 'securities',
                                           'stock_market', 'ths_code', 'tdx_code'])
            return Market(code, m[0], m[1], m[2], code + '.' + m[4], m[4] + code)


'''获取通信达终端的板块信息'''


def get_stock_block(blockname, type='default', tdx_path='D:/通达信金融终端', returnSeries=True):
    if type == 'zs':
        block_file = resolve(tdx_path, 'T0002', 'hq_cache', 'block_zs.dat')
    elif type == 'fg':
        block_file = resolve(tdx_path, 'T0002', 'hq_cache', 'block_fg.dat')
    elif type == 'gn':
        block_file = resolve(tdx_path, 'T0002', 'hq_cache', 'block_gn.dat')
    else:
        block_file = resolve(tdx_path, 'T0002', 'hq_cache', 'block.dat')
    bf = BlockReader().get_df(block_file, BlockReader_TYPE_GROUP)
    if bf is None:
        return bf
    else:
        if blockname == None:
            dateList = bf['code_list'].values[0].split(',')
        else:
            if isinstance(blockname, str):
                dateList = bf.query('blockname == @blockname')['code_list'].values[0].split(',')
            elif isinstance(blockname, (tuple, list)):
                if isinstance(blockname, tuple):
                    blockname = list(blockname)
                dateList = ','.join(bf.query('blockname in ' + str(blockname))['code_list'].values).split(',')
            else:
                dateList = bf['code_list'].values[0].split(',')
        if returnSeries:
            return pd.Series(dateList)
        else:
            return dateList

def get_stock_cycle(rule):
    if rule in ['1m', '1M', '1t', '1T', 8]:
        category = 8
        cycle = '1T'
        bars = 240
        minute = 1
        compression = 1
    elif rule in ['5m', '5M', '5t', '5T', 0]:
        category = 0
        cycle = '5T'
        bars = 48
        minute = 5
        compression = 5
    elif rule in ['15m', '15M', '15t', '15T', 1]:
        category = 1
        cycle = '15T'
        bars = 16
        minute = 15
        compression = 15
    elif rule in ['30m', '30M', '30t', '30T', 2]:
        category = 2
        cycle = '30T'
        bars = 8
        minute = 30
        compression = 30
    elif rule in ['60m', '60M', '60t', '60T', '1h', '1H', 3]:
        category = 3
        cycle = '60T'
        bars = 4
        minute = 60
        compression = 60
    elif rule in ['1d', '1D', 'Day', 4]:
        category = 4
        cycle = '1D'
        bars = 1
        minute = 240
        compression = 1
    elif rule in ['1w', '1W', 'Week', 5]:
        category = 5
        cycle = '1W'
        bars = 0
        minute = 1200
        compression = 1
    elif rule in ['month', 'Month', 'MM', 6]:
        category = 6
        cycle = '1M'
        bars = 0
        minute = 4800
        compression = 1
    elif rule in ['quarter', 'Quarter', '1Q', '1q', 10]:
        category = 10
        cycle = '1Q'
        bars = 0
        minute = 14400
        compression = 1
    elif rule in ['year', 'Year', '1Y', '1q', 11]:
        category = 11
        cycle = '1Y'
        bars = 0
        minute = 60000
        compression = 1
    else:
        category = 9
        cycle = '1D'
        bars = 1
        minute = 240
        compression = 1
    if cycle in ['1T', '5T', '15T', '30T', '60T']:
        timeframe = bt.TimeFrame.Minutes
    elif cycle == '1D':
        timeframe = bt.TimeFrame.Days
    elif cycle == '1W':
        timeframe = bt.TimeFrame.Weeks
    elif cycle == '1M':
        timeframe = bt.TimeFrame.Months
    elif cycle == '1Y':
        timeframe = bt.TimeFrame.Years
    else:
        timeframe = bt.TimeFrame.Days
    Cycle = namedtuple('Cycle', ['category', 'cycle', 'bars', 'timeframe', 'minute', 'compression'])
    return Cycle(category, cycle, bars, timeframe, minute, compression)

class TdxStore:
    def __init__(self, config):
        platform = self.checkPlatform()
        if platform == 'Windows':
            self.TdxPath = config.Source.tdx.win_path
        elif platform in ['MacOS', 'Linux', 'Windows Subsystem for Linux (WSL)']:
            self.TdxPath = config.Source.tdx.mac_path
        else:
            pass

    def checkPlatform(self):
        if sys.platform.startswith('win'):
            return 'Windows'
        elif sys.platform.startswith('darwin'):
            return 'MacOS'
        elif sys.platform.startswith('linux'):
            if 'Microsoft' in platform.release(): \
                    return 'Windows Subsystem for Linux (WSL)'
            else:
                return 'Linux'
        else:
            return 'Unknown'

    def getStockBlock(self):
        api = TdxHq_API()
        return api.get_and_parse_block_info(TDXParams.BLOCK_SZ)

    """批量股票的本地和网络数据取出"""
    def liveResampleDatas(self, stocks, days=None):
        api = TdxHq_API()
        stock = dict()
        with api.connect("119.147.212.81", 7709):
            try:
                local = self.getLocalStockBars(stocks)
                for index, code in enumerate(stocks):
                    info = get_stock_market(code[0])
                    rule = get_stock_cycle(code[1])
                    max_bar = 800 if rule.bars * 10 > 800 else rule.bars * 10 + 1
                    network = api.to_df(api.get_security_bars(rule.category, info.market, info.stock_code, 0, max_bar))
                    df_data = pd.DataFrame()
                    if not network.empty:
                        df_data = df_data.assign(datetime=network['datetime'])
                        df_data = df_data.assign(open=network['open'])
                        df_data = df_data.assign(high=network['high'])
                        df_data = df_data.assign(low=network['low'])
                        df_data = df_data.assign(close=network['close'])
                        df_data = df_data.assign(volume=network['vol'])
                        df_data = df_data.assign(ratio=network['close'].diff())
                    stock_data = pd.concat([local[index], df_data], ignore_index=True)
                    stock_data = stock_data.dropna()
                    stock_data = stock_data.drop_duplicates(subset=['datetime'], keep='first', ignore_index=True)
                    stock_data.index = pd.to_datetime(stock_data.datetime)
                    stock[info.tdx_code + '_' + rule.cycle] = self._return_stock_data(stock_data, info, rule, days)
            except Exception as e:
                print(e)
        return stock

    def liveResampleData(self, code, cycle=None, days=None):
        api = TdxHq_API()
        rule = get_stock_cycle(cycle if cycle is not None else '1D')
        max_bar = 800 if rule.bars * 10 > 800 else rule.bars * 10 + 1
        with api.connect("119.147.212.81", 7709):
            local = self.getLocalStockBar(code, rule.cycle)
            info = get_stock_market(code)
            network = api.to_df(api.get_security_bars(rule.category, info.market, code, 0, max_bar))
            df_data = pd.DataFrame()
            df_data = df_data.assign(datetime=network['datetime'])
            df_data = df_data.assign(open=network['open'])
            df_data = df_data.assign(high=network['high'])
            df_data = df_data.assign(low=network['low'])
            df_data = df_data.assign(close=network['close'])
            df_data = df_data.assign(volume=network['vol'])
            df_data = df_data.assign(ratio=network['close'].diff())
            stock_data = pd.concat([local, df_data], ignore_index=True)
            stock_data = stock_data.dropna()
            stock_data = stock_data.drop_duplicates(subset=['datetime'], keep='first', ignore_index=True)
            stock_data.index = pd.to_datetime(stock_data.datetime)
            return self._return_stock_data(stock_data, info, rule, days)


    """单一个股的本地和网络数据合并取出"""
    def get_stock_data(self, code, cycle=None, days=None):
        info = get_stock_market(code)
        rule = get_stock_cycle(cycle if cycle is not None else '1D')
        local = self.getStockBarsFromLocal(info.ths_code, cycle)
        max_bar = 800 if rule.bars * 10 > 800 else rule.bars * 10 + 1
        if cycle in ['1d', '1D', 'd', 'D']:
            network = self.getStockBarsFromApi(4, code, 0, max_bar, True)
        elif cycle in ['1m', '1M', '1t', '1T']:
            network = self.getStockBarsFromApi(8, code, 0, max_bar, True)
        elif cycle in ['5m', '5M', '5t', '5T']:
            network = self.getStockBarsFromApi(0, code, 0, max_bar, True)
        elif cycle in ['15m', '15M', '15t', '15T']:
            network = self.getStockBarsFromApi(1, code, 0, max_bar, True)
        elif cycle in ['30m', '30M', '30t', '30T']:
            network = self.getStockBarsFromApi(2, code, 0, max_bar, True)
        elif cycle in ['60m', '60M', '60t', '60T', '1h', '1H', 'h', 'H']:
            network = self.getStockBarsFromApi(3, code, 0, max_bar, True)
        else:
            network = pd.DataFrame()
        stock_data = pd.concat([local, network], ignore_index=True)
        stock_data = stock_data.drop_duplicates(subset=['datetime'], keep='first', ignore_index=True)
        stock_data.index = pd.to_datetime(stock_data.datetime)
        return self._return_stock_data(stock_data, info, rule, days)

    """数据的统一返回格式"""
    def _return_stock_data(self, stock_data, info, rule, days=None):
        if days is not None:
            now = datetime.datetime.now()
            start = (now - datetime.timedelta(days=days)).strftime('%Y-%m-%d')
            stock_data = stock_data.loc[stock_data['datetime'] > start]
        else:
            if rule.cycle in ['1T', '5T', '15T', '30T', '60T']:
                stock_data = stock_data.loc[stock_data['datetime'] >= '2020-11-01 09:30']
        Stock = namedtuple('Stock', ['data', 'length', 'columns', 'info', 'cycle', 'start', 'end'])
        return Stock(stock_data, stock_data.shape[0], stock_data.columns, info, rule,
                     stock_data['datetime'][0], stock_data["datetime"].tail(1).values[0])

    '''
    获取股票k线数据
    * category-> K线种类
        0 5分钟K线 
        1 15分钟K线 
        2 30分钟K线 
        3 1小时K线 
        4 日K线
        5 周K线
        6 月K线
        7 1分钟
        8 1分钟K线 9 日K线
        10 季K线
        11 年K线
    * market -> 市场代码 0:深圳，1:上海
    * stockcode -> 证券代码;
    * start -> 指定的范围开始位置;
    * count -> 用户要请求的 K 线数目，最大值为 800
    '''

    def _stuck_bf_data(self, data):
        df_data = pd.DataFrame()
        df_data = df_data.assign(datetime=data['datetime'])
        df_data = df_data.assign(open=data['open'])
        df_data = df_data.assign(high=data['high'])
        df_data = df_data.assign(low=data['low'])
        df_data = df_data.assign(close=data['close'])
        df_data = df_data.assign(volume=data['vol'])
        df_data = df_data.assign(ratio=data['close'].diff())
        df_data = df_data.dropna()  # 去除空值
        df_data.index = pd.to_datetime(df_data.datetime)
        return df_data

    def getStockBarsFromApi(self, stocks, start, count, is_dp=False):
        api = TdxHq_API()
        with api.connect("119.147.212.81", 7709):
            return_data = list()
            for stock in stocks:
                info = get_stock_market(stock[0])
                rule = get_stock_cycle(stock[1])
                if is_dp:
                    data = api.to_df(api.get_security_bars(rule.category, info.market, info.stock_code, start, count))
                    df_data = self._stuck_bf_data(data)
                    return_data.append(self._return_stock_data(df_data, info, rule))
                else:
                    return_data.append(api.get_security_bars(category, info.market, code, start, count))
            return return_data

    def getStockBarFromApi(self, category, code, start, count, is_dp=False):
        info = get_stock_market(code)
        rule = get_stock_cycle(category)
        api = TdxHq_API()
        if api.connect("119.147.212.81", 7709):
            if is_dp:
                data = api.to_df(api.get_security_bars(rule.category, info.market, code, start, count))
                df_data = self._stuck_bf_data(data)
                return self._return_stock_data(df_data, info, rule)
            else:
                return api.get_security_bars(rule.category, info.market, code, start, count)

    '''
    获取股票指数数据
    * category-> K线种类
        0 5分钟K线 
        1 15分钟K线 
        2 30分钟K线 
        3 1小时K线 
        4 日K线
        5 周K线
        6 月K线
        7 1分钟
        8 1分钟K线 
        9 日K线
        10 季K线
        11 年K线
    * market -> 市场代码 0:深圳，1:上海
    * stockcode -> 证券代码;
    * start -> 指定的范围开始位置;
    * count -> 用户要请求的 K 线数目，最大值为 800
    '''

    def getIndexBarsFromApi(self, category, code, start, count, isDp=False):
        info = get_stock_market(code)
        ips = [(v[1], v[2]) for v in hq_hosts]
        random.shuffle(ips)  # 获取5个随机ip作为ip池
        ips5 = ips[:5]
        ippool = AvailableIPPool(TdxHq_API, ips5)  ## IP 池对象
        primary_ip, hot_backup_ip = ippool.sync_get_top_n(2)  ## 选出M, H
        ## 生成hqpool对象，第一个参数为TdxHq_API后者 TdxExHq_API里的一个，第二个参数为ip池对象。
        api = TdxHqPool_API(TdxHq_API, ippool)
        ## connect 函数的参数为M, H 两组 (ip, port) 元组
        with api.connect(primary_ip, hot_backup_ip):
            if (isDp):
                return api.to_df(api.get_index_bars(category, info.market, code, start, count))
            else:
                return api.get_index_bars(category, info.market, code, start, count)

    def singleResample(self, code, param, days=None):
        stock_rule = get_stock_cycle(param.trend_cycle)
        stock_data = self.get_stock_data(code, stock_rule.cycle).data
        stock_data = self._overlyingIndicators(stock_data, param)
        return self._return_stock_data(stock_data, get_stock_market(code), stock_rule, days)

    def doubleResample(self, code, param, days=None):
        stock_rule = get_stock_cycle(param.stock_cycle)
        trend_rule = get_stock_cycle(param.trend_cycle)
        stock_data = self.get_stock_data(code, stock_rule.cycle).data
        columns = stock_data.columns.values
        multi = int(trend_rule.minute / stock_rule.minute)
        rounding = int(len(stock_data) % multi)
        surplus = int(len(stock_data) // multi)
        idx = list(range(0, surplus)) * multi
        if rounding > 0:
            for i in range(0, rounding):
                idx.append(surplus)
        idx_new = idx.copy()
        idx_new.pop(0)
        idx_new = [x - 1 for x in idx_new]
        idx_new.append(surplus)
        idx.sort()
        idx_new.sort()
        stock_data['idx'] = idx
        trend_data = stock_data.groupby(by='idx').agg({'datetime': 'last', 'open': 'first', 'high': 'max', 'low': 'min',
                                             'close': 'last', 'volume': 'sum', 'idx': 'last'}).dropna()
        stock_data['idx'] = idx_new
        trend_data = trend_data.assign(ratio=trend_data['close'].diff())
        trend_data.index = pd.to_datetime(trend_data.datetime)
        trend_data = self._overlyingIndicators(trend_data, param)
        temp_series = stock_data.iloc[-1]
        stock_data = pd.merge(stock_data, trend_data, how='right', on='idx', suffixes=('', '_trend'))
        stock_data = stock_data.drop(['idx', 'datetime_trend'], axis=1, inplace=False)
        stock_data.loc[len(stock_data)-1, columns] = temp_series
        stock_data.index = pd.to_datetime(stock_data.datetime)
        return self._return_stock_data(stock_data, get_stock_market(code), stock_rule, days)

    """
    叠加技术指标
    """
    def _overlyingIndicators(self, data, param):
        if 'volume_period' in dir(param):
            data = data.assign(volume_ma=ta.MA(data.volume, timeperiod=param.volume_period))
        if 'ema_s_period' in dir(param):
            data = data.assign(ema_s=ta.EMA(data.close, param.ema_s_period))
        if 'ema_m_period' in dir(param):
            data = data.assign(ema_m=ta.EMA(data.close, param.ema_m_period))
        if 'ema_l_period' in dir(param):
            data = data.assign(ema_l=ta.EMA(data.close, param.ema_l_period))
        if 'boll_period' in dir(param):
            factor = param.boll_period if 'boll_factor' in dir(param) else 2
            upper, middle, lower = ta.BBANDS(data.close, timeperiod=param.boll_period,
                                             nbdevup=factor, nbdevdn=factor, matype=0)
            data = data.assign(upper=round(upper, 3))
            data = data.assign(middle=round(middle, 3))
            data = data.assign(lower=round(lower, 3))
            data = data.assign(bb=round((data.close - lower) / (upper - lower) * 100, 2))
            data = data.assign(width=round((upper - lower) / middle * 100, 2))
        if 'macd_short' in dir(param):
            long = param.macd_long if 'macd_me2' in param else 26
            signal = param.macd_sig if 'macd_sig' in param else 9
            dif, dea, macd = ta.MACD(data.close, param.macd_short, long, signal)
            data = data.assign(dif=dif)
            data = data.assign(dea=dea)
            data = data.assign(macd=macd)
        if 'atr_period' in dir(param):
            multiplier = param.atr_multiple if 'atr_multiple' in param else 1.5
            data = data.assign(atr=ta.ATR(data.high, data.low, data.close, timeperiod=param.atr_period))
            atr_lower = data.apply(lambda x: x.low - x.atr * multiplier, axis=1)
            data = data.drop(['atr'], axis=1, inplace=False)
            data = data.assign(atr_loss=atr_lower.rolling(window=param.atr_period).max())
        if 'platform_period' in dir(param):
            data = data.assign(platform_upper=round(data.high.rolling(param.platform_period).max(), 3))
            data = data.assign(platform_lower=round(data.low.rolling(param.platform_period).min(), 3))
            data = data.assign(stop_surplus=round(data.close.rolling(param.platform_period).max(), 3))
        if 'morphology' in dir(param) and param.morphology:
            data = data.assign(
                CDLINVERTEDHAMMER=ta.CDLINVERTEDHAMMER(data.open, data.high, data.low, data.close))
            data = data.assign(
                CDLSHOOTINGSTAR=ta.CDLSHOOTINGSTAR(data.open, data.high, data.low, data.close))
            data = data.assign(
                CDLDARKCLOUDCOVER=ta.CDLDARKCLOUDCOVER(data.open, data.high, data.low, data.close))
            data = data.assign(
                CDLEVENINGSTAR=ta.CDLEVENINGSTAR(data.open, data.high, data.low, data.close))
            data = data.assign(
                CDLHANGINGMAN=ta.CDLHANGINGMAN(data.open, data.high, data.low, data.close))
            data = data.assign(
                CDLHAMMER=ta.CDLHAMMER(data.open, data.high, data.low, data.close))
            data = data.assign(
                CDLINVERTEDHAMMER=ta.CDLINVERTEDHAMMER(data.open, data.high, data.low, data.close))
            data = data.assign(
                CDLMORNINGDOJISTAR=ta.CDLMORNINGDOJISTAR(data.open, data.high, data.low, data.close))
            data = data.assign(
                CDLUNIQUE3RIVER=ta.CDLUNIQUE3RIVER(data.open, data.high, data.low, data.close))
            data = data.assign(
                CDL3BLACKCROWS=ta.CDL3BLACKCROWS(data.open, data.high, data.low, data.close))
            '''morphology_rise = list()
            morphology_drop = list()
            if data['high'] == data['platform_high_trend']:
                if ta.CDLINVERTEDHAMMER(data.open, data.high, data.low, data.close) > 0:
                    morphology_rise.append({
                        'title': '上升趋势倒锤头',
                        'explain': '上影线较长，长度为实体2倍以上，无下影线，在上涨趋势顶部看跌'})
                if ta.CDLSHOOTINGSTAR(data.open, data.high, data.low, data.close):
                    morphology_rise.append({
                        'title': '射击之星',
                        'explain': '上影线至少为实体长度的2倍，同时没有下影线，在上涨趋势顶部看跌'})
                if ta.CDLDARKCLOUDCOVER(data.open, data.high, data.low, data.close):
                    morphology_rise.append({
                        'title': '乌云压顶',
                        'explain': '第一日长阳，第二日开盘价高于前一日最高价，收盘价处于前一日实体中部以下，预示下跌'})
                if ta.CDLEVENINGSTAR(data.open, data.high, data.low, data.close):
                    morphology_rise.append({
                        'title': '黄昏之星',
                        'explain': '上升趋势中，第一日阳线，第二日价格振幅较小，第三日阴线，预示顶部反转'})
                if ta.CDLHANGINGMAN(data.open, data.high, data.low, data.close):
                    morphology_rise.append({
                        'title': '上吊线',
                        'explain': '形状与锤子类似，处于上升趋势的顶部，预示着趋势反转'})
            if data.low == data.platform_low_trend:
                if ta.CDLHAMMER(data.open, data.high, data.low, data.close):
                    morphology_drop.append({
                        'title': '下降趋势锤头线',
                        'explain': '实体较短，无上影线，下影线大于实体长度两倍，处于下跌趋势底部，预示反转'})
                if ta.CDLINVERTEDHAMMER(data.open, data.high, data.low, data.close):
                    morphology_drop.append({
                        'title': '下降趋势倒锤头',
                        'explain': '实上影线较长，长度为实体2倍以上，无下影线，在下跌趋势底部，预示着趋势反转'})
            if ta.CDLMORNINGDOJISTAR(data.open, data.high, data.low, data.close, penetration=0):
                morphology_drop.append({
                    'title': '十字晨星',
                    'explain': '三日K线模式，基本模式为晨星，第二日K线为十字星，预示底部反转'})
            if ta.CDLUNIQUE3RIVER(data.open, data.high, data.low, data.close):
                morphology_drop.append({
                    'title': '独特三河',
                    'explain': '三日K线模式，下跌趋势中，第一日长阴线，第二日为锤头，最低价创新低，第三日开盘价低于第二日收盘价，收阳线，收盘价不高于第二日收盘价，预示着反转'})
            if ta.CDL3BLACKCROWS(data.open, data.high, data.low, data.close):
                morphology_rise.append({
                    'title': '三只乌鸦',
                    'explain': '连续三根阴线，而且每日收盘价都下跌且接近最低价，开盘价都在上根K线实体内，预示下跌'})
            trend_data = trend_data.assign(morphology_rise_trend=morphology_rise)
            trend_data = trend_data.assign(morphology_drop_trend=morphology_drop)'''
        return data

    # 根据通达信5分钟周期数据，生成其他周期数据
    def lc5Resample(self, data_lc5, rule):
        if rule == '60T':
            idx = list(range(0, len(data_lc5) // 12)) * 12
            idx.sort()
            data_lc5['idx'] = idx
            df_data = data_lc5.groupby(by='idx').agg({
                'datetime': 'last', 'open': 'first', 'high': 'max',
                'low': 'min', 'close': 'last', 'volume': 'sum'}).dropna()
        else:
            df_data = data_lc5.resample(rule, closed='right', label='right').agg({
                'datetime': 'last', 'open': 'first', 'high': 'max',
                'low': 'min', 'close': 'last', 'volume': 'sum'}).dropna()
        df_data = df_data.assign(ratio=df_data['close'].diff())
        df_data.index = pd.to_datetime(df_data.datetime)
        return df_data

    def _stuck_param_func(self, code, cycle=None):
        info = get_stock_market(code)
        if cycle.upper() in ['1H', '60M', '60T', '30M', '30T', '15m', '15T', '5M', '5T']:
            filename = info.ths_code.split(".")[1] + info.ths_code.split(".")[0] + ".lc5"
            filepath = resolve(self.TdxPath, "vipdoc", info.ths_code.split(".")[1], "fzline", filename)
            pack = 'HHffffllf'
        elif cycle.upper() in ['1M', '1T']:
            filename = info.ths_code.split(".")[1] + info.ths_code.split(".")[0] + ".lc1"
            filepath = resolve(self.TdxPath, "vipdoc", info.ths_code.split(".")[1], "minline", filename)
            pack = 'HHffffllf'
        else:
            filename = info.ths_code.split(".")[1] + info.ths_code.split(".")[0] + ".day"
            filepath = resolve(self.TdxPath, "vipdoc", info.ths_code.split(".")[1], "lday", filename)
            pack = 'lllllfll'
        return filepath, pack, cycle.upper()

    def _read_file_func(self, path, pack, cycle):
        kline = self._read_kline(path, pack)
        kline.index = pd.to_datetime(kline["datetime"])
        kline.index.name = "index"
        usecols = ("datetime open high low close ratio volume".split())
        if cycle.upper() in ['1H', '60M', '60T', '30M', '30T', '15m', '15T']:
            return self.lc5Resample(kline[usecols], rule=cycle)
        else:
            return kline[usecols]

    def getLocalStockBars(self, codes):
        params = list()
        for code in codes:
            params.append(self._stuck_param_func(code[0], code[1]))
        pool = Pool(5)
        return pool.starmap(self._read_file_func, params)

    # 获取本地K线数据
    def getLocalStockBar(self, code, cycle=None):
        filename, pack, filepath = self._stuck_param_func(code, cycle)
        return self._read_file_func(filename, pack, filepath)

    # 根据二进制前两段拿到日期分时
    def _parseDateTime(self, h1, h2) -> str:  # H1->0,1字节; H2->2,3字节;
        return self._parseDate(h1) + " " + self._parseTime(h2)

    def _formatDate(self, year, m, d):
        if (m < 10):
            month = '0' + str(m)
        else:
            month = str(m)
        if (d < 10):
            day = '0' + str(d)
        else:
            day = str(d)
        return str(year) + "-" + month + "-" + day

    def _parseDate(self, h):
        year = math.floor(h / 2048) + 2004  # 解析出年
        month = math.floor(h % 2048 / 100)  # 月
        day = h % 2048 % 100  # 日
        return self._formatDate(year, month, day)

    def _parseTime(self, h):
        hour = math.floor(h / 60)  # 小时
        minute = h % 60  # 分钟
        if hour < 10:  # 如果小时小于两位, 补0
            hour = "0" + str(hour)
        if minute < 10:  # 如果分钟小于两位, 补0
            minute = "0" + str(minute)
        return str(hour) + ":" + str(minute)

    # 读取K线源文件
    def _read_kline(self, filepath, pack='lllllfll'):
        with open(filepath, "rb") as f:
            usecols = "datetime open high low close ratio volume".split()
            buffers = []
            i = 0
            while True:
                buffer = f.read(32)
                if not buffer:
                    break
                buffer = struct.unpack(pack, buffer)
                if i == 0:
                    preClose = buffer[4] / 100
                ratio = round((buffer[4] / 100 - preClose) / preClose * 100, 3)
                preClose = buffer[4] / 100
                i = i + 1
                if (pack == 'lllllfll'):
                    year = int(buffer[0] / 10000)
                    month = int((buffer[0] % 10000) / 100)
                    day = (buffer[0] % 10000) % 100
                    date = self._formatDate(year, month, day)
                    buffers.append((date + ' 15:00', buffer[1] / 100, buffer[2] / 100, buffer[3] / 100,
                                    buffer[4] / 100, ratio, buffer[6]))
                else:
                    buffers.append((self._parseDateTime(buffer[0], buffer[1]),
                                    buffer[2], buffer[3], buffer[4], buffer[5], ratio, buffer[7]))
            kline = pd.DataFrame(buffers, columns=usecols)
        kline["datetime"] = kline["datetime"].astype(str)
        price_columns = ["open", "high", "low", "close"]
        kline[price_columns] = kline[price_columns].apply(lambda x: x)
        return kline